package com.sqllog.admin.sql.entity;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.sqllog.admin.cache.CanalClientCache;
import com.sqllog.admin.sql.domain.ClientRunThreadException;
import com.sqllog.admin.sql.mapper.CanalClientConfigInfoMapper;
import com.sqllog.admin.utils.SpringContextUtils;
import com.sqllog.config.CanalClientConfig;
import com.sqllog.core.CanalClient;
import com.sqllog.core.SimpleCanalClient;
import com.sqllog.listener.SimpleCanalEventListener;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * sql 监听实例
 *
 * @author chennengcheng
 * @date 2022-01-20 15:52
 * @remark
 */

@Slf4j
@Getter
@Setter
public class CanalClientConfigInfo implements Serializable {
    
    private String id;

    private String title;
    
    private String instanceName;
    
    private String clusterEnabled;
    
    private String host;
    
    private int port;
    
    private String userName;
    
    private String password;
    
    private int batchSize;

    private String filter;

    private int consumerThreadSize;

    private String zookeeperAddress;

    private String delFlag;

    /**
     * 0: 未启动  1: 正常  2: 启动中  3: 连接异常
     */
    private String status;

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date createDate;

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date updateDate;

    private String errorLog;

    public static CanalClientConfig getCanalClientConfig(CanalClientConfigInfo canalClientConfigInfo){
        CanalClientConfig canalServerInstance = new CanalClientConfig();
        canalServerInstance.setHost(canalClientConfigInfo.getHost());
        canalServerInstance.setPort(canalClientConfigInfo.getPort());
        canalServerInstance.setBatchSize(canalClientConfigInfo.getBatchSize());
        canalServerInstance.setClusterEnabled(canalClientConfigInfo.getClusterEnabled().equals("1"));
        canalServerInstance.setUserName(canalClientConfigInfo.getUserName());
        canalServerInstance.setPassword(canalClientConfigInfo.getPassword());
        canalServerInstance.setInstanceName(canalClientConfigInfo.getInstanceName());
        canalServerInstance.setFilter(canalClientConfigInfo.getFilter());
        String zookeeperAddress = canalClientConfigInfo.getZookeeperAddress();
        if (StringUtils.isNotBlank(zookeeperAddress)) {
            String[] addressList = zookeeperAddress.split(",");
            Set<String> addressSet = Arrays.stream(addressList).collect(Collectors.toSet());
            canalServerInstance.setZookeeperAddress(addressSet);
        }

        return canalServerInstance;
    }

    public static CanalClient startCanalClient(CanalClientConfigInfo canalClientConfigInfo,
                                        List<SimpleCanalEventListener> simpleCanalEventListeners){
        CanalClientConfig canalClientConfig = CanalClientConfigInfo.getCanalClientConfig(canalClientConfigInfo);
        CanalClient canalClient = new SimpleCanalClient(canalClientConfig, simpleCanalEventListeners);
        CanalClientConfigInfoMapper canalClientConfigInfoMapper = SpringContextUtils.getBean(CanalClientConfigInfoMapper.class);
        try {
            canalClientConfigInfoMapper.updateStatusById("2", canalClientConfigInfo.getId());
            canalClient.start(new ClientRunThreadException());
            canalClientConfigInfoMapper.updateStatusById("1", canalClientConfigInfo.getId());
        }catch (Exception e) {
            log.error("启动异常");
            canalClientConfigInfoMapper.updateStatusById("3", canalClientConfigInfo.getId());
        }

        CanalClientCache.canalClientContext.put(canalClientConfigInfo.getId(), canalClient);
        return canalClient;
    }

    public static CanalClient stopCanalClient(CanalClientConfigInfo canalClientConfigInfo){
        CanalClientConfigInfoMapper canalClientConfigInfoMapper = SpringContextUtils.getBean(CanalClientConfigInfoMapper.class);
        CanalClient canalClient = CanalClientCache.canalClientContext.get(canalClientConfigInfo.getId());
        canalClient.stop();
        canalClientConfigInfoMapper.updateStatusById("0", canalClientConfigInfo.getId());
//        CanalClientCache.canalClientContext.remove(canalClientConfigInfo.getId(), canalClient);
        return canalClient;
    }
}
